package org.rency.mq.activemq.listener.factory;

import org.rency.mq.activemq.constant.DestinationType;
import org.rency.mq.activemq.jms.support.impl.DestinationResolverImpl;
import org.rency.mq.activemq.listener.handler.MessageHandler;
import org.rency.mq.activemq.listener.handler.impl.AbstractMessageHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public abstract class AbstractMessageListenerContainer extends org.springframework.jms.listener.DefaultMessageListenerContainer
		implements MessageListenerContainer {
	private static Logger log = LoggerFactory.getLogger(AbstractMessageListenerContainer.class);

	private MessageHandler messageHandler;
    private boolean sessionAware = false;
	protected void validateConfiguration() {
		if (isSubscriptionDurable() && !isPubSubDomain()) {
			throw new IllegalArgumentException(
					"A durable subscription requires a topic (pub-sub domain)");
		}

		if (messageHandler == null && getMessageListener() == null) {
			throw new IllegalArgumentException(
					"MessageHandler and MessageListener cannot be null at same time");
		}

		if (isSubscriptionDurable() && getConcurrentConsumers() != 1) {
			throw new IllegalArgumentException(
					"Only 1 concurrent consumer supported for durable subscription");
		}
	}

	public void initialize() {
		if (getDestinationName() == null){
			if (messageHandler != null && messageHandler instanceof AbstractMessageHandler) {
				AbstractMessageHandler handler = (AbstractMessageHandler) messageHandler;
				String destination = handler.getDestination();
				
				if(destination != null){
					int destinationType = handler.getDestinationType();
					
					if(destinationType != DestinationType.QUEUE && destinationType != DestinationType.TOPIC){
						throw new IllegalArgumentException("Invalid destination type:" + destinationType);
					}
					
					setDestinationName(destination);
					setPubSubDomain(destinationType == DestinationType.TOPIC);
				}

				if (getDestinationName() == null) {
					throw new NullPointerException("Destination cann't be null.");
				}
			}
		}
			
		if (messageHandler != null && getMessageListener() == null) {
			initMessageListener();
		}

		setDestinationResolver(DestinationResolverImpl.getInstance());

		super.initialize();
	}

	public abstract void initMessageListener();

	public int getContainerStatus() {
		if (!isActive()) {
			return MQMessageListenerStatus.INACTIVE.getValue();
		}
		if (isRunning()) {
			return MQMessageListenerStatus.RUNNING.getValue();
		} else {
			return MQMessageListenerStatus.STOPPED.getValue();
		}
	}

	public void startContainer() {
		new Thread(
				new Runnable() {
					
					public void run() {
						while(true){
							try {
								if(getContainerStatus() == MQMessageListenerStatus.INACTIVE.getValue()){
									AbstractMessageListenerContainer.super.afterPropertiesSet();
								}else if(getContainerStatus() == MQMessageListenerStatus.STOPPED.getValue()){
									AbstractMessageListenerContainer.super.start();
									log.info("Listener on {} start successfully.",getDestinationName());
									if(log.isInfoEnabled()){

									}
								}
								else{
									log.info("Listener on {} is already started.",getDestinationName());
									break;
								}
							} catch (Exception e) {
								log.warn("Init the listener failed, and application will retry later.", e);
							}
						}
					}
				}
		).start();
	}

	public void stopContainer() {
		if (this.getContainerStatus() == MQMessageListenerStatus.RUNNING
				.getValue()) {
			if (log.isInfoEnabled()) {
				log.info("Stopping listener on {} ...",getDestinationName());
			}

			super.stop();
		}
	}

	public MessageHandler getMessageHandler() {
		return messageHandler;
	}

	public void setMessageHandler(MessageHandler messageHandler) {
		if(messageHandler == null){
			throw new IllegalArgumentException("Message handler cannot be null.");
		}
		
		this.messageHandler = messageHandler;
	}
    public boolean isSessionAware() {
        return sessionAware;
    }

    public void setSessionAware(boolean sessionAware) {
        this.sessionAware = sessionAware;
    }
}

